package com.hexinfo.dmpro.monitor.service;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor;
import com.alibaba.druid.sql.parser.SQLParserUtils;
import com.alibaba.druid.sql.parser.SQLStatementParser;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.hexinfo.dmpro.common.base.BaseService;
import com.hexinfo.dmpro.common.base.Page;
import com.hexinfo.dmpro.common.base.PageResult;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.monitor.dao.CmStatementAuditDao;
import com.hexinfo.dmpro.monitor.dao.CmSwitchDao;
import com.hexinfo.dmpro.monitor.model.CmStatementAudit.*;
import com.hexinfo.dmpro.monitor.service.serviceparam.CmTargetParamService;
import com.hexinfo.dmpro.monitor.service.serviceparam.GiveAnAlarmService;
import com.hexinfo.dmpro.monitor.util.ConnMysqlUtil;
import com.hexinfo.dmpro.monitor.util.ConvertUtil;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.sql.Clob;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;

@Slf4j
@Service("cmStatementAuditService")
public class CmStatementAuditService extends BaseService {

    public static final String GREATER_THAN = ">";
    public static final String LESS_THAN = "<";

    public static final String UNIT_TIME_S = "s";
    public static final String UNIT_TIME_MIN = "min";
    public static final String UNIT_TIME_H = "h";

    public static final String UNIT_memory_KIB = "kib";
    public static final String UNIT_memory_MIB = "mib";
    public static final String UNIT_memory_GIB = "gib";

    @Value("${warn.publishTheme}")
    private String publishTheme;
    @Value("${warn.yarnPublishTheme}")
    private String yarnPublishTheme;
    @Value("${warn.impalaException}")
    private String impalaException;
    @Value("${database.impalaTableName}")
    private String impalaTableName;
    @Value("${database.yarnTableName}")
    private String yarnTableName;
    @Value("${database.impalaSqlPath}")
    private String impalaSqlPath;
    @Value("${database.yarnSqlPath}")
    private String yarnSqlPath;

    private static final Logger logger = LoggerFactory.getLogger(GiveAnAlarmService.class);
    @Resource
    private CmStatementAuditDao statementAuditDao;
    @Resource
    private CmSwitchService cmSwitchService;
    @Autowired
    private ConnMysqlUtil connMysqlUtil;
    @Autowired
    private CmTargetParamService cmTargetParamService;

    public List<Map<String, String>> selectClusterName() {
        return statementAuditDao.selectClusterName();
    }

    public Map<String, Object> selectAuditAlarm(ComponentMonthModel model) {
        Map<String, Object> resultMap = new HashMap<>();
        List<Map<String, String>> allClusterList = statementAuditDao.selectAllCluster();
        try {
            List<Map<String, Object>> resultList = new ArrayList<>();
            List<String> timeList = new ArrayList<>();
            List<String> nameList = new ArrayList<>();
            for (Map<String, String> clusterMap : allClusterList) {
                String cmFlag = clusterMap.get("cmFlag");
                String clusterFlag = clusterMap.get("clusterFlag");
                String cmName = clusterMap.get("cmName");
                String clusterName = clusterMap.get("clusterName");
                nameList.add(cmName + "--" + clusterName);
                List<Map<String, String>> list = new ArrayList<>();
                model.setCmFlag(cmFlag);
                model.setClusterFlag(clusterFlag);
                if ("IMPALA".equals(model.getType())) {
                    ComponentMonthModel monthModel = getTableList(model,impalaTableName);
                    list = statementAuditDao.selectImpalaAuditAlarm(monthModel);
                }
                if ("YARN".equals(model.getType())) {
                    ComponentMonthModel monthModel = getTableList(model,yarnTableName);
                    list = statementAuditDao.selectYarnAuditAlarm(monthModel);
                }
                Map<String, Object> map = dealWithData(list, model.getStartTime(), model.getEndTime());
                List<Map<String, String>> dealWithMapList = (List<Map<String, String>>) map.get("dataList");
                timeList = (List<String>) map.get("time");
                List<String> numberList = new ArrayList<>();
                for (Map<String, String> map1 : dealWithMapList) {
                    numberList.add(map1.get("number"));
                }
                Map<String, Object> map2 = new HashMap<>();
                map2.put("name", cmName + "--" + clusterName);
                map2.put("type", "line");
                map2.put("data", numberList);
                resultList.add(map2);
            }
            resultMap.put("series", resultList);
            resultMap.put("timeList", timeList);
            resultMap.put("nameList", nameList);
        } catch (Exception e) {
            log.error("组件审计汇总查询异常：{}" + e);
        } finally {
//            DynamicDataSourceContextHolder.clear();
        }
        return resultMap;
    }

    public Map<String, Object> selectComponentDetails(ComponentMonthModel model, Page page) {
        PageResult pageResult = null;
        if ("IMPALA".equals(model.getType())) {
            ComponentMonthModel monthModel = getTableList(model,impalaTableName);
            pageResult = selectPage(statementAuditDao, "selectImpalaComponentDetails", page, monthModel);
        } else if ("YARN".equals(model.getType())) {
            ComponentMonthModel monthModel = getTableList(model,yarnTableName);
            pageResult = selectPage(statementAuditDao, "selectYarnComponentDetails", page, monthModel);
        }
        Map<String, Object> result = new HashMap<>(2);
        result.put("content", pageResult.getContent());
        result.put("totalElements", pageResult.getTotalElements());
        return result;
    }

    public List<String> selectModelNameByCluster(String component) {
        return statementAuditDao.selectModelNameByCluster(component);
    }

    public Map<String, Object> selectModelSummary(ComponentMonthModel model) {
        Map<String, Object> resultMap = new HashMap<>();
        List<Map<String, String>> allClusterList = statementAuditDao.selectAllCluster();
        try {
            List<Map<String, Object>> resultList = new ArrayList<>();
            List<String> timeList = new ArrayList<>();
            List<String> nameList = new ArrayList<>();
            for (Map<String, String> clusterMap : allClusterList) {
                String cmFlag = clusterMap.get("cmFlag");
                String clusterFlag = clusterMap.get("clusterFlag");
                String cmName = clusterMap.get("cmName");
                String clusterName = clusterMap.get("clusterName");
                nameList.add(cmName + "--" + clusterName);
                List<Map<String, String>> list = new ArrayList<>();
                model.setCmFlag(cmFlag);
                model.setClusterFlag(clusterFlag);
                if ("IMPALA".equals(model.getType())) {
                    ComponentMonthModel monthModel = getTableList(model,impalaTableName);
                    list = statementAuditDao.selectImpalaModelSummary(monthModel);
                }
                if ("YARN".equals(model.getType())) {
                    ComponentMonthModel monthModel = getTableList(model,yarnTableName);
                    list = statementAuditDao.selectYarnModelSummary(monthModel);
                }
                Map<String, Object> map = dealWithData(list, model.getStartTime(), model.getEndTime());
                List<Map<String, String>> dealWithMapList = (List<Map<String, String>>) map.get("dataList");
                timeList = (List<String>) map.get("time");
                List<String> numberList = new ArrayList<>();
                for (Map<String, String> map1 : dealWithMapList) {
                    numberList.add(map1.get("number"));
                }
                Map<String, Object> map2 = new HashMap<>();
                map2.put("name", cmName + "--" + clusterName);
                map2.put("type", "line");
                map2.put("data", numberList);
                resultList.add(map2);
            }
            resultMap.put("series", resultList);
            resultMap.put("timeList", timeList);
            resultMap.put("nameList", nameList);
        } catch (Exception e) {
            log.error("模型审计汇总查询异常：{}" + e);
        } finally {
//            DynamicDataSourceContextHolder.clear();
        }
        return resultMap;
    }

    public Map<String, Object> selectModelDetails(ComponentMonthModel model, Page page) {
        PageResult pageResult = null;
        if ("IMPALA".equals(model.getType())) {
            ComponentMonthModel monthModel = getTableList(model,impalaTableName);
            pageResult = selectPage(statementAuditDao, "selectImpalaModelDetails", page, monthModel);
        } else if ("YARN".equals(model.getType())) {
            ComponentMonthModel monthModel = getTableList(model,yarnTableName);
            pageResult = selectPage(statementAuditDao, "selectYarnModelDetails", page, monthModel);
        }
        Map<String, Object> result = new HashMap<>(2);
        result.put("content", pageResult.getContent());
        result.put("totalElements", pageResult.getTotalElements());
        return result;
    }

    public Map<String, Object> dealWithData(List<Map<String, String>> list, String startTime, String endTime) throws ParseException {
        List<Map<String, String>> dataList = new ArrayList<>();
        Calendar cale = Calendar.getInstance();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        // 获取当前日期
        System.out.println(sdf.format(cale.getTime()));
        // 获取一个月前的日期
        cale.add(Calendar.MONTH, -1);
        Date startDate = sdf.parse(startTime);
        Date endDate = sdf.parse(endTime);
        List<String> timeList = new ArrayList<>();
        while (!startDate.equals(endDate)) {
            timeList.add(sdf.format(startDate));
            cale.setTime(startDate);
            cale.add(Calendar.DATE, 1);
            startDate = cale.getTime();
        }
        timeList.add(sdf.format(endDate));
        for (String s : timeList) {
            boolean b = false;
            String value = null;
            for (Map<String, String> map : list) {
                if (map.get("time").equals(s)) {
                    b = true;
                }
                if (b) {
                    value = String.valueOf(map.get("count"));
                    break;
                }
            }
            if (!b) {
                Map<String, String> maps = new HashMap<>();
                maps.put("time", s);
                maps.put("number", "0");
                dataList.add(maps);
            } else {
                Map<String, String> maps = new HashMap<>();
                maps.put("time", s);
                maps.put("number", value);
                dataList.add(maps);
            }
        }
        Map<String, Object> map = new HashMap<>();
        map.put("dataList", dataList);
        map.put("time", timeList);
        return map;
    }

    public Map<String, Object> findConfigModel(ConfigureModel model, Page page) {
        PageResult pageResult = selectPage(statementAuditDao, "findConfigModel", page, model);
        Map<String, Object> result = new HashMap<>(2);
        result.put("content", pageResult.getContent());
        result.put("totalElements", pageResult.getTotalElements());
        return result;
    }

    /**
     * 数据插入模型配置
     *
     * @param model
     */
    public void insertConfigModel(ConfigureModel model) {
        log.info("数据插入模型配置:model:" + model);
        List<ConfigureModel> configureModels = statementAuditDao.selectModelIsExists(model);
        if(configureModels.size() > 0) {
            throw new RuntimeException("模型名称不允许重复!");
        }
        statementAuditDao.insertConfigModel(model);
    }

    public void updateConfigModel(ConfigureModel model) {
        statementAuditDao.updateConfigModel(model);
    }

    public void deleteConfigModel(String id) {
        statementAuditDao.deleteConfigModel(id);
    }

    //Impala月份区间，连表查询
    public List<ImpalaParameterModel> selectImpalaParamMonth(ComponentMonthModel model) {
        ComponentMonthModel monthModel = getTableList(model,impalaTableName);
        return statementAuditDao.selectImpalaParamMonth(monthModel);
    }

    public PageResult selectImpalaParamMonth(ComponentMonthModel model,Page page) {
        ComponentMonthModel monthModel = getTableList(model,impalaTableName);
        PageResult selectImpalaParamMonth = selectPage(statementAuditDao, "selectImpalaParamMonth", page, monthModel);
        return selectImpalaParamMonth;
    }
    /**
     * 获取时间范围内的所有相关月份表
     * @param model
     * @return
     */
    public ComponentMonthModel getTableList(ComponentMonthModel model,String tableName) {
        List<String> stringList = connMysqlUtil.getMonthBetweenDate(model.getStartTime(), model.getEndTime());
        model.setTableName(tableName + "_" + stringList.get(0));
        if (stringList.size()>1){
            List<String> tableNameList = new ArrayList<>();
            for (int i=1;i<stringList.size();i++) {
                String name = tableName + "_" + stringList.get(i);
                tableNameList.add(name);
            }
            model.setTableNameList(tableNameList);
        }
        return model;
    }

    //Yarn月份区间，连表查询
    public List<YarnParameterModel> selectYarnParamMonth(ComponentMonthModel model) {
        ComponentMonthModel monthModel = getTableList(model,yarnTableName);
        return statementAuditDao.selectYarnParamMonth(monthModel);
    }

    public PageResult selectYarnParamMonth(ComponentMonthModel model,Page page) {
        ComponentMonthModel monthModel = getTableList(model,yarnTableName);
        PageResult selectYarnParamMonth = selectPage(statementAuditDao,"selectYarnParamMonth",page,monthModel);
        return selectYarnParamMonth;
    }

    public List exportImpalaExcel(ComponentMonthModel model){
        List<ImpalaParameterModel> models = this.selectImpalaParamMonth(model);
        List list = new ArrayList<>();
        String[] titleRow = new String[]{"父集群名称","子集群名称","模型名称","用户","数据库","SQL语句","查询类型",
        "查询状态","持续时间","生成的行","读取的 HDFS 字节","规划等待时间","线程：CPU 时间","线程：总时间"};
        list.add(titleRow);
        if (models.size() > 0) {
            for (int i = 0; i < models.size(); i++) {
                String[] arr = new String[14];
                arr[0] = models.get(i).getCmName();
                arr[1] = models.get(i).getClusterName();
                arr[2] = models.get(i).getModelName();
                arr[3] = models.get(i).getUser();
                arr[4] = models.get(i).getDatabase();
                arr[5] = models.get(i).getStatement();
                arr[6] = models.get(i).getQueryType();
                arr[7] = models.get(i).getQueryState();
                arr[8] = models.get(i).getQueryDuration();
                arr[9] = models.get(i).getRowsProduced();
                arr[10] = models.get(i).getHdfsBytesRead();
                arr[11] = models.get(i).getPlanningWaitTime();
                arr[12] = models.get(i).getThreadCpuTime();
                arr[13] = models.get(i).getThreadTotalTime();
                list.add(arr);
            }
        }
        return list;
    }

    public List exportYarnExcel(ComponentMonthModel model){
        List<YarnParameterModel> models = this.selectYarnParamMonth(model);
        List list = new ArrayList<>();
        String[] titleRow = new String[]{"父集群名称","子集群名称","模型名称","名称","池","用户","应用程序状态",
        "持续时间","已分配内存的秒数","已分配 VCore 的秒数"," 写入的 HDFS 字节","CPU 时间","读取的文件字节"};
        list.add(titleRow);
        if (models.size() > 0) {
            for (int i = 0; i < models.size(); i++) {
                String[] arr = new String[14];
                arr[0] = models.get(i).getCmName();
                arr[1] = models.get(i).getClusterName();
                arr[2] = models.get(i).getModelName();
                arr[3] = models.get(i).getName();
                arr[4] = models.get(i).getPool();
                arr[5] = models.get(i).getUser();
                arr[6] = models.get(i).getState();
                arr[7] = models.get(i).getElapsedTime();
                arr[8] = models.get(i).getAllocatedMemorySeconds();
                arr[9] = models.get(i).getAllocatedVcoreSeconds();
                arr[10] = models.get(i).getHdfsBytesWritten();
                arr[11] = models.get(i).getCpuMilliseconds();
                arr[12] = models.get(i).getFileBytesRead();
                list.add(arr);
            }
        }
        return list;
    }

    /**
     * 定时任务 每10分钟执行一次
     */
    public void getImpalaParameterData() {
        try {
            log.info("获取impala语句参数开始*************************");
            String startDate = LocalDateTime.now().plusHours(1).minusMinutes(Long.parseLong("10")).toString().substring(0, 19).replace("T", " ");
            //获取及整理，平台上配置的CM配置信息
            List<Map<String, String>> monitorReqList = cmSwitchService.getCmConfigInfo("", "", "impala");
            for (Map<String, String> map : monitorReqList) {
                getImpalaParam(map,startDate);
            }
        } catch (Exception e) {
            log.error("获取impala语句参数异常1：{}" + e);
        }
    }

    //按照配置信息获取数据，保存到数据库中
    public void getImpalaParam(Map<String, String> map, String startDate){
        try {
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/impalaQueries?from={startTime}";
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmFlag = map.get("cmFlag");
            String cmName = map.get("cmName");
            String clusterFlag = map.get("clusterFlag");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            log.info("cmFlag:" + cmFlag);
            log.info("clusterFlag:" + clusterFlag);
            log.info("serviceName" + serviceName);
            //时间为当前时间前10分钟
            String iso86Time = getISO86Time(startDate);
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName)
                    .replace("{startTime}", iso86Time);
            //按配置信息获取cm数据
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                JSONObject obj = JSONObject.parseObject(objectMap.get("body").toString());
                JSONArray jsonArray = obj.getJSONArray("queries");
                List<ImpalaParameterModel> list = new ArrayList<>();
                //cm数据分类筛选插入数据库
                if (jsonArray != null && jsonArray.size() > 0) {
                    List<ConfigureModel> queryModels = statementAuditDao.selectModelConfig("QUERY","FINISHED");
                    List<ConfigureModel> ddlModels = statementAuditDao.selectModelConfig("DDL","FINISHED");
                    List<ConfigureModel> dmlModels = statementAuditDao.selectModelConfig("DML","FINISHED");
                    List<ConfigureModel> queryModelsError = statementAuditDao.selectModelConfig("QUERY","EXCEPTION");
                    List<ConfigureModel> ddlModelsError = statementAuditDao.selectModelConfig("DDL","EXCEPTION");
                    List<ConfigureModel> dmlModelsError = statementAuditDao.selectModelConfig("DML","EXCEPTION");
                    List<ConfigureModel> naModelsError = statementAuditDao.selectModelConfig("N/A","EXCEPTION");
                    List<ConfigureModel> queryModelsPending = statementAuditDao.selectModelConfig("QUERY","PENDING");
                    List<ConfigureModel> ddlModelsPending = statementAuditDao.selectModelConfig("DDL","PENDING");
                    List<ConfigureModel> dmlModelsPending = statementAuditDao.selectModelConfig("DML","PENDING");
                    //获取等待中任务数量
                    int queryCount = 0;
                    int ddlCount = 0;
                    int dmlCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        if ("QUERY".equals(parseObject.get("queryType").toString())) {
                            if("select 1".equals(parseObject.get("statement").toString().replace("\n", " "))){
                                continue;
                            }
                            if("FINISHED".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : queryModels) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaQuery(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            if("EXCEPTION".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : queryModelsError) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaQuery(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            //running前的3个状态，任务等待中状态
                            if("CREATED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "INITIALIZED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "COMPILED".equalsIgnoreCase(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : queryModelsPending) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaQuery(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                        queryCount++;
                                    }
                                }
                            }
                        }
                        if ("DDL".equals(parseObject.get("queryType").toString())) {
                            if("FINISHED".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : ddlModels) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDdl(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            if("EXCEPTION".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : ddlModelsError) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDdl(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            //running前的3个状态，任务等待中状态
                            if("CREATED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "INITIALIZED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "COMPILED".equalsIgnoreCase(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : ddlModelsPending) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDdl(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                        ddlCount++;
                                    }
                                }
                            }
                        }
                        if ("DML".equals(parseObject.get("queryType").toString())) {
                            if("FINISHED".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : dmlModels) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDml(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            if("EXCEPTION".equals(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : dmlModelsError) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDml(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                    }
                                }
                            }
                            //running前的3个状态，任务等待中状态
                            if("CREATED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "INITIALIZED".equalsIgnoreCase(parseObject.get("queryState").toString()) ||
                                    "COMPILED".equalsIgnoreCase(parseObject.get("queryState").toString())) {
                                for (ConfigureModel model : dmlModelsPending) {
                                    ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                    insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                    String modelName = validateImpalaDml(parameterModel, model);
                                    if (StringUtils.isNotBlank(modelName)) {
                                        parameterModel.setModelName(modelName);
                                        list.add(parameterModel);
                                        dmlCount++;
                                    }
                                }
                            }
                        }
                        if ("N/A".equals(parseObject.get("queryType").toString())) {
                            for (ConfigureModel model : naModelsError) {
                                ImpalaParameterModel parameterModel = new ImpalaParameterModel();
                                insertImpalaPara(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                String modelName = validateImpalaNa(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                    }
                    //impala任务失败告警
                    int exceptionCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        if("EXCEPTION".equals(parseObject.get("queryState").toString())) {
                            exceptionCount++;
                        }
                    }
                    if(exceptionCount > 0){
                        String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+impalaException+"： "+ exceptionCount;
                        cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,impalaException);
                    }
                    System.out.println("4444444444444444444444444444444444444444444444444444");
                    log.info("list" + list);
                    if (list.size() > 0) {
                        //insertImpalaParameterData(list);
                        insertMonthImpalaParameterData(list);
                        //等待任务数量告警
                        if (queryCount > 0 || ddlCount > 0 || dmlCount > 0){
                            int count = queryCount+ddlCount+dmlCount;
                            String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+publishTheme+"：QUERY类型任务等待数 "+
                                    queryCount+"、DDL类型任务等待数 "+ddlCount+"、DML类型任务等待数 "+dmlCount+"，总等待数 "+count;
                            cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,publishTheme);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取impala语句参数异常2:" + e);
        }
    }

    public void insertImpalaPara(String cmFlag, String cmName, String clusterFlag, String clusterName,
                                                 ImpalaParameterModel parameterModel, JSONObject parseObject) throws ParseException{
        parameterModel.setCmFlag(cmFlag);
        parameterModel.setCmName(cmName);
        parameterModel.setClusterFlag(clusterFlag);
        parameterModel.setClusterName(clusterName);
        parameterModel.setQueryId(parseObject.getString("queryId"));
        parameterModel.setUser(parseObject.get("user").toString());
        parameterModel.setDatabase(parseObject.get("database").toString());
        parameterModel.setStatement(parseObject.get("statement").toString().replace("\n", " "));
        parameterModel.setCoordinatorHostId(parseObject.getJSONObject("coordinator").getString("hostId"));
        parameterModel.setQueryState(parseObject.get("queryState").toString());
        parameterModel.setQueryDuration(parseObject.get("durationMillis").toString());
        parameterModel.setRowsProduced(parseObject.getString("rowsProduced"));
        parameterModel.setQueryType(parseObject.get("queryType").toString());
        Map parseObject1 = JSON.parseObject(String.valueOf(parseObject.getJSONObject("attributes")), Map.class);
//        System.out.println(parseObject.getJSONObject("attributes"));
        //查询impala任务同类语句并发数，的查询条件
//        String impalaTableInfo = getImpalaTableInfo(parseObject.get("statement").toString().replace("\n", " "));
//        parseObject1.put("tableInfo", impalaTableInfo);
        parameterModel.setOtherParameterData(JSON.toJSONString(parseObject1));

//        parameterModel.setOtherParameterData(parseObject.getJSONObject("attributes").toString());
        parameterModel.setStartTime(getBjTime(parseObject.getString("startTime")));
        Date dat = new Date();
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        if (parseObject.get("endTime") != null) {
            parameterModel.setEndTime(getBjTime(parseObject.getString("endTime")));
        }else if (StringUtils.isBlank(parseObject.get("durationMillis").toString())){
            long elapsedTime = getElapsedTime(parameterModel.getStartTime(), sf.format(dat));
            parameterModel.setQueryDuration(String.valueOf(elapsedTime));
        }
        String date = sf.format(dat);
        parameterModel.setCreateTime(date);
    }

    public String validateImpalaQuery(ImpalaParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //内存不足
        String a1 = (String) parseObject.get("oom");
        if (StringUtils.isNotBlank(a1) && "null" != a1) {
            String oom = parseObject.get("oom").toString();
            String memory = map.get("memory").toString();
            if (StringUtils.isNotBlank(memory) && StringUtils.isNotBlank(oom)) {
                if (!memory.equals(oom)) {
                    return "";
                }
            }
        }

        //缺少统计数据
        String a2 = (String) parseObject.get("stats_missing");
        if (StringUtils.isNotBlank(a2) && "null" != a2) {
            String statsMissing = parseObject.get("stats_missing").toString();
            String statisticalData = map.get("statisticalData").toString();
            if (StringUtils.isNotBlank(statisticalData) && StringUtils.isNotBlank(statsMissing)) {
                if (!statisticalData.equals(statsMissing)) {
                    return "";
                }
            }
        }
        //持续时间
        String a3 = String.valueOf(parameterModel.getQueryDuration());
        if (StringUtils.isNotBlank(a3) && "null" != a3 && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(a3);
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        //线程：CPU 时间
        String s2 = (String) parseObject.get("thread_cpu_time");
        if (StringUtils.isNotBlank(s2) && "null" != s2 && StringUtils.isNotBlank(map.get("thread").toString())) {
            long threadCpuTime = Long.valueOf(s2);
            long thread = Long.valueOf(map.get("thread").toString());
            String threadUnit = map.get("threadUnit").toString();
            String threadSymbol = map.get("threadSymbol").toString();
            boolean b2 = validateTime(threadCpuTime, thread, threadUnit, threadSymbol);
            if (!b2) {
                return "";
            }
        }
        //读取的 HDFS 字节
        String s3 = (String) parseObject.get("hdfs_bytes_read");
        if (StringUtils.isNotBlank(s3) && "null" != s3 && StringUtils.isNotBlank(map.get("byte").toString())) {
            long hdfsBytesRead = Long.valueOf(s3);
            long aByte = Long.valueOf(map.get("byte").toString());
            String byteUnit = map.get("byteUnit").toString();
            String byteSymbol = map.get("byteSymbol").toString();
            boolean b3 = validateByte(hdfsBytesRead, aByte, byteUnit, byteSymbol);
            if (!b3) {
                return "";
            }
        }
        //客户端获取等待时间
        String s4 = (String) parseObject.get("client_fetch_wait_time");
        if (StringUtils.isNotBlank(s4) && "null" != s4 && StringUtils.isNotBlank(map.get("wait").toString())) {
            long clientFetchWaitTime = Long.valueOf(s4);
            long wait = Long.valueOf(map.get("wait").toString());
            String waitUnit = map.get("waitUnit").toString();
            String waitSymbol = map.get("waitSymbol").toString();
            boolean b4 = validateTime(clientFetchWaitTime, wait, waitUnit, waitSymbol);
            if (!b4) {
                return "";
            }
        }
        //生成的行
        String s5 = String.valueOf(parameterModel.getRowsProduced());
        if (StringUtils.isNotBlank(s5) && "null" != s5 && StringUtils.isNotBlank(map.get("rows").toString())) {
            long rowsProduced = Long.valueOf(s5);
            long rows = Long.valueOf(map.get("rows").toString());
            String rowsSymb = map.get("rowsSymbol").toString();
            boolean b5 = validateValue(rowsProduced, rows, rowsSymb);
            if (!b5) {
                return "";
            }
        }
        //规划等待时间百分比
        String s6 = (String) parseObject.get("planning_wait_time_percentage");
        if (StringUtils.isNotBlank(s6) && "null" != s6 && StringUtils.isNotBlank(map.get("percentage").toString())) {
            long i = Long.valueOf(s6);
            long percentage = Long.valueOf(map.get("percentage").toString());
            String percentageSymbol = map.get("percentageSymbol").toString();
            boolean b6 = validateValue(i, percentage, percentageSymbol);
            if (!b6) {
                return "";
            }
        }
        //累积内存使用峰值
        String s7 = (String) parseObject.get("memory_aggregate_peak");
        if (StringUtils.isNotBlank(s7) && "null" != s7 && StringUtils.isNotBlank(map.get("peakValue").toString())) {
            long integer = parseScientificNotationToLong(s7);
            long peakValue = Long.valueOf(map.get("peakValue").toString());
            String peakValueUnit = map.get("peakValueUnit").toString();
            String peakValueSymbol = map.get("peakValueSymbol").toString();
            boolean b7 = validateByte(integer, peakValue, peakValueUnit, peakValueSymbol);
            if (!b7) {
                return "";
            }
        }
        return model.getModelName();
    }

    public String validateImpalaDdl(ImpalaParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //内存不足
        String a1 = (String) parseObject.get("oom");
        if (StringUtils.isNotBlank(a1) && "null" != a1) {
            String oom = a1;
            String memory = map.get("memory").toString();
            if (StringUtils.isNotBlank(memory) && StringUtils.isNotBlank(oom)) {
                if (!memory.equals(oom)) {
                    return "";
                }
            }
        }
        //缺少统计数据
        String a2 = (String) parseObject.get("stats_missing");
        if (StringUtils.isNotBlank(a2) && "null" != a2) {
            String statsMissing = a2;
            String statisticalData = map.get("statisticalData").toString();
            if (StringUtils.isNotBlank(statisticalData) && StringUtils.isNotBlank(statsMissing)) {
                if (!statisticalData.equals(statsMissing)) {
                    return "";
                }
            }
        }
        //持续时间
        String a3 = String.valueOf(parameterModel.getQueryDuration());
        if (StringUtils.isNotBlank(a3) && "null" != a3 && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(a3);
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        //客户端获取等待时间
        String s4 = (String) parseObject.get("client_fetch_wait_time");
        if (StringUtils.isNotBlank(s4) && "null" != s4 && StringUtils.isNotBlank(map.get("wait").toString())) {
            long clientFetchWaitTime = Long.valueOf(s4);
            long wait = Long.valueOf(map.get("wait").toString());
            String waitUnit = map.get("waitUnit").toString();
            String waitSymbol = map.get("waitSymbol").toString();
            boolean b4 = validateTime(clientFetchWaitTime, wait, waitUnit, waitSymbol);
            if (!b4) {
                return "";
            }
        }
        //规划等待时间
        String s6 = (String) parseObject.get("planning_wait_time");
        if (StringUtils.isNotBlank(s6) && "null" != s6 && StringUtils.isNotBlank(map.get("waitTime").toString())) {
            long i = Long.valueOf(s6);
            long percentage = Long.valueOf(map.get("waitTime").toString());
            String waitTimeUnit = map.get("waitTimeUnit").toString();
            String percentageSymbol = map.get("waitTimeSymbol").toString();
            boolean b6 = validateTime(i, percentage, waitTimeUnit, percentageSymbol);
            if (!b6) {
                return "";
            }
        }

        return model.getModelName();
    }

    public String validateImpalaDml(ImpalaParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //内存不足
        String a1 = (String) parseObject.get("oom");
        if (StringUtils.isNotBlank(a1) && "null" != a1) {
            String oom = a1;
            String memory = map.get("memory").toString();
            if (StringUtils.isNotBlank(memory) && StringUtils.isNotBlank(oom)) {
                if (!memory.equals(oom)) {
                    return "";
                }
            }
        }
        //缺少统计数据
        String a2 = (String) parseObject.get("stats_missing");
        if (StringUtils.isNotBlank(a2) && "null" != a2) {
            String statsMissing = a2;
            String statisticalData = map.get("statisticalData").toString();
            if (StringUtils.isNotBlank(statisticalData) && StringUtils.isNotBlank(statsMissing)) {
                if (!statisticalData.equals(statsMissing)) {
                    return "";
                }
            }
        }
        //生成的行
        String s5 = String.valueOf(parameterModel.getRowsProduced());
        if (StringUtils.isNotBlank(s5) && "null" != s5 && StringUtils.isNotBlank(map.get("rows").toString())) {
            long rowsProduced = Long.valueOf(s5);
            long rows = Long.valueOf(map.get("rows").toString());
            String rowsSymb = map.get("rowsSymbol").toString();
            boolean b5 = validateValue(rowsProduced, rows, rowsSymb);
            if (!b5) {
                return "";
            }
        }
        //持续时间
        String a3 = String.valueOf(parameterModel.getQueryDuration());
        if (StringUtils.isNotBlank(a3) && "null" != a3 && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(a3);
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        //规划等待时间
        String s6 = (String) parseObject.get("planning_wait_time");
        if (StringUtils.isNotBlank(s6) && "null" != s6 && StringUtils.isNotBlank(map.get("waitTime").toString())) {
            long i = Long.valueOf(s6);
            long percentage = Long.valueOf(map.get("waitTime").toString());
            String waitTimeUnit = map.get("waitTimeUnit").toString();
            String percentageSymbol = map.get("waitTimeSymbol").toString();
            boolean b6 = validateTime(i, percentage, waitTimeUnit, percentageSymbol);
            if (!b6) {
                return "";
            }
        }
        //线程：CPU 时间
        String s2 = (String) parseObject.get("thread_cpu_time");
        if (StringUtils.isNotBlank(s2) && "null" != s2 && StringUtils.isNotBlank(map.get("thread").toString())) {
            long threadCpuTime = Long.valueOf(s2);
            long thread = Long.valueOf(map.get("thread").toString());
            String threadUnit = map.get("threadUnit").toString();
            String threadSymbol = map.get("threadSymbol").toString();
            boolean b2 = validateTime(threadCpuTime, thread, threadUnit, threadSymbol);
            if (!b2) {
                return "";
            }
        }
        return model.getModelName();
    }

    public String validateImpalaNa(ImpalaParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //内存不足
        String a1 = (String) parseObject.get("oom");
        if (StringUtils.isNotBlank(a1) && "null" != a1) {
            String oom = a1;
            String memory = map.get("memory").toString();
            if (StringUtils.isNotBlank(memory) && StringUtils.isNotBlank(oom)) {
                if (!memory.equals(oom)) {
                    return "";
                }
            }
        }
        //缺少统计数据
        String a2 = (String) parseObject.get("stats_missing");
        if (StringUtils.isNotBlank(a2) && "null" != a2) {
            String statsMissing = a2;
            String statisticalData = map.get("statisticalData").toString();
            if (StringUtils.isNotBlank(statisticalData) && StringUtils.isNotBlank(statsMissing)) {
                if (!statisticalData.equals(statsMissing)) {
                    return "";
                }
            }
        }
        //持续时间
        String a3 = String.valueOf(parameterModel.getQueryDuration());
        if (StringUtils.isNotBlank(a3) && "null" != a3 && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(a3);
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        return model.getModelName();
    }

    /**
     * 科学计算法转long类型
     * @param input
     * @return
     */
    public long parseScientificNotationToLong(String input) {
        double num = Double.parseDouble(input); // 转换为 double 类型
        return (long) num; // 转换为 long 类型
    }

    public boolean validateTime(long sourceValue, long modelValue, String modelUnit, String modelSymbol) {
        boolean b = false;
        switch (modelUnit) {
            case UNIT_TIME_S:
                modelValue = modelValue * 1000;
                break;
            case UNIT_TIME_MIN:
                modelValue = modelValue * 60 * 1000;
                break;
            case UNIT_TIME_H:
                modelValue = modelValue * 60 * 60 * 1000;
            default:
                break;
        }
        if (modelSymbol.equals(GREATER_THAN)) {
            if (sourceValue > modelValue) {
                b = true;
            }
        }
        if (modelSymbol.equals(LESS_THAN)) {
            if (sourceValue < modelValue) {
                b = true;
            }
        }
        return b;
    }

    public boolean validateByte(long sourceValue, long modelValue, String modelUnit, String modelSymbol) {
        boolean b = false;
        switch (modelUnit) {
            case UNIT_memory_KIB:
                modelValue = modelValue * 1024;
            case UNIT_memory_MIB:
                modelValue = modelValue * 1024 * 1024;
                break;
            case UNIT_memory_GIB:
                modelValue = modelValue * 1024 * 1024 * 1024;
                break;
            default:
                break;
        }
        if (modelSymbol.equals(GREATER_THAN)) {
            if (sourceValue > modelValue) {
                b = true;
            }
        }
        if (modelSymbol.equals(LESS_THAN)) {
            if (sourceValue < modelValue) {
                b = true;
            }
        }
        return b;
    }

    public boolean validateValue(long sourceValue, long modelValue, String modelSymbol) {
        boolean b = false;
        if (modelSymbol.equals(GREATER_THAN)) {
            if (sourceValue > modelValue) {
                b = true;
            }
        }
        if (modelSymbol.equals(LESS_THAN)) {
            if (sourceValue < modelValue) {
                b = true;
            }
        }
        return b;
    }

    //Impala语句审计保留历史记录（2年）数据表按照月份表
    public void insertMonthImpalaParameterData(List<ImpalaParameterModel> list) {
        try {
            int pointsDataLimit = 200;
            int listSize = list.size();
            int maxSize = listSize - 1;
            List<ImpalaParameterModel> newList = new ArrayList<>();
            //获取当前月表名
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMM");
            String name = impalaTableName + "_" + sdf.format(new Date());
            logger.info("当前月表名："+name);
            for (int i = 0; i < listSize; i++) {
                newList.add(list.get(i));
                //一次性插入数据最多200条
                if (pointsDataLimit == newList.size() || i == maxSize) {
                    ImpalaParameterMonthModel imp = new ImpalaParameterMonthModel();
                    imp.setImpalaList(newList);
                    imp.setTableName(name);
                    statementAuditDao.insertMonthImpalaParameterData(imp);
                    newList.clear();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("impala语句任务参数插入impala表异常：{}" + e);
        }
    }

    /**
     * 定时任务 每10分钟执行一次
     */
    public void getYarnParameterData() {
        try {
            String startDate = LocalDateTime.now().plusHours(1).minusMinutes(Long.parseLong("10")).toString().substring(0, 19).replace("T", " ");
            List<Map<String, String>> monitorReqList = cmSwitchService.getCmConfigInfo("", "", "yarn");
            for (Map<String, String> map : monitorReqList) {
                getMapreduceParam(map, startDate);
                getFlinkParam(map, startDate);
                getSparkParam(map, startDate);
                getTezParam(map, startDate);
            }
        } catch (Exception e) {
            log.error("获取yarn语句mapreduce任务参数异常：{}" + e);
            e.printStackTrace();
        }
    }

    public void getMapreduceParam(Map<String, String> map, String startDate) {
        try {
            log.info("获取Mapreduce参数开始***************************************");
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/yarnApplications?filter=(application_type=MAPREDUCE)&from={startTime}";
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmFlag = map.get("cmFlag");
            String cmName = map.get("cmName");
            String clusterFlag = map.get("clusterFlag");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            //时间为当前时间前10分钟
            String iso86Time = getISO86Time(startDate);
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName)
                    .replace("{startTime}", iso86Time);
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                JSONObject obj = JSONObject.parseObject(objectMap.get("body").toString());
                JSONArray jsonArray = obj.getJSONArray("applications");
                List<YarnParameterModel> list = new ArrayList<>();
                if (jsonArray != null && jsonArray.size() > 0) {
                    List<ConfigureModel> mapreduceModels = statementAuditDao.selectModelConfig("MAPREDUCE","SUCCEEDED");
                    List<ConfigureModel> mapreduceModelsError = statementAuditDao.selectModelConfig("MAPREDUCE","FAILED");
                    List<ConfigureModel> mapreduceModelsPending = statementAuditDao.selectModelConfig("MAPREDUCE","PENDING");
                    //获取等待中任务数量
                    int mapreduceCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        if("SUCCEEDED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel mapreduceModel : mapreduceModels) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyMapreduce(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                String modelName = validateMapreduceModelName(parameterModel, mapreduceModel);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("FAILED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel mapreduceModel : mapreduceModelsError) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyMapreduce(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                String modelName = validateMapreduceModelName(parameterModel, mapreduceModel);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("NEW".equalsIgnoreCase(parseObject.getString("state")) ||
                                "NEW_SAVING".equalsIgnoreCase(parseObject.getString("state")) ||
                                "SUBMITTED".equalsIgnoreCase(parseObject.getString("state")) ||
                                "ACCEPTED".equalsIgnoreCase(parseObject.getString("state"))) {
                            for (ConfigureModel mapreduceModel : mapreduceModelsPending) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyMapreduce(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject);
                                String modelName = validateMapreduceModelName(parameterModel, mapreduceModel);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                    mapreduceCount++;
                                }
                            }
                        }
                    }
                    log.info("mapreduce list大小:" + list.size());
                    if (list.size() > 0) {
                        insertMonthYarnParameterData(list);
                        //等待任务数量告警
                        if (mapreduceCount > 0){
                            String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+yarnPublishTheme+"：MAPREDUCE任务等待数 "+ mapreduceCount;
                            //微信、钉钉告警，然后添加进告警日志表
                            cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,yarnPublishTheme);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取yarn语句mapreduce任务参数异常：{}" + e);
            e.printStackTrace();
        }
    }

    public void copyMapreduce(String cmFlag, String cmName, String clusterFlag, String clusterName,
                              YarnParameterModel parameterModel, JSONObject parseObject) throws ParseException {
        parameterModel.setCmFlag(cmFlag);
        parameterModel.setCmName(cmName);
        parameterModel.setClusterFlag(clusterFlag);
        parameterModel.setClusterName(clusterName);
        parameterModel.setApplicationId(parseObject.getString("applicationId"));
        parameterModel.setApplicationType("MAPREDUCE");
        if (ObjectUtil.isNotEmpty(parseObject.getJSONObject("attributes").get("hive_query_string"))){
            if (StrUtil.isNotBlank(parseObject.getJSONObject("attributes").get("hive_query_string").toString())){
                parameterModel.setName(parseObject.getJSONObject("attributes").get("hive_query_string").toString().replace("\n", " "));
            }
        }else {
            parameterModel.setName(parseObject.get("name").toString().replace("\n", " "));
        }
        parameterModel.setPool(parseObject.getString("pool"));
        parameterModel.setUser(parseObject.getString("user"));
        parameterModel.setState(parseObject.getString("state"));
        if(ObjectUtil.isNotEmpty(parseObject.get("allocatedMemorySeconds"))){
            parameterModel.setAllocatedMemorySeconds(parseObject.get("allocatedMemorySeconds").toString());
        }
        if(ObjectUtil.isNotEmpty(parseObject.get("allocatedVcoreSeconds"))){
            parameterModel.setAllocatedVcoreSeconds(parseObject.get("allocatedVcoreSeconds").toString());
        }
        Map parseObject1 = JSON.parseObject(String.valueOf(parseObject.getJSONObject("attributes")), Map.class);
        //查询mapreduce任务同类语句并发数，的查询条件
//        String mapreduceTableInfo = getMapreduceTableInfo(parseObject.getJSONObject("attributes").get("hive_query_string").toString().replace("\n", " "));
//        parseObject1.put("tableInfo", mapreduceTableInfo);
        parameterModel.setOtherParameterData(JSON.toJSONString(parseObject1));
        parameterModel.setStartTime(getBjTime(parseObject.getString("startTime")));
        Date dat = new Date();
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        if (parseObject.get("endTime") != null) {
            parameterModel.setEndTime(getBjTime(parseObject.getString("endTime")));
            long elapsedTime = getElapsedTime(parameterModel.getStartTime(), parameterModel.getEndTime());
            parameterModel.setElapsedTime(String.valueOf(elapsedTime));
        } else {
            long elapsedTime = getElapsedTime(parameterModel.getStartTime(), sf.format(dat));
            parameterModel.setElapsedTime(String.valueOf(elapsedTime));
        }
        String date = sf.format(dat);
        parameterModel.setCreateTime(date);
    }

    public long getElapsedTime(String startTime, String endTime) throws ParseException {
        long time;
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date start = format.parse(startTime);
        Date end = format.parse(endTime);
        time = (end.getTime() - start.getTime()) / 1000;
        return time;
    }

    public String validateMapreduceModelName(YarnParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //持续时间
        if (StringUtils.isNotBlank(parameterModel.getElapsedTime()) && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(parameterModel.getElapsedTime());
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        //CPU 时间
        String a2 = (String) parseObject.get("cpu_milliseconds");
        if (StringUtils.isNotBlank(a2) && "null" != a2 && StringUtils.isNotBlank(map.get("cpu").toString())) {
            long cpuMilliseconds = Long.valueOf(a2);
            long cpu = Long.valueOf(map.get("cpu").toString());
            String cpuUnit = map.get("cpuUnit").toString();
            String cpuSymbol = map.get("cpuSymbol").toString();
            boolean b2 = validateTime(cpuMilliseconds, cpu, cpuUnit, cpuSymbol);
            if (!b2) {
                return "";
            }
        }
        //时隙
        String a3 = (String) parseObject.get("slots_millis");
        if (StringUtils.isNotBlank(a3) && "null" != a3 && StringUtils.isNotBlank(map.get("timeSlot").toString())) {
            long slotsMillis = Long.valueOf(a3);
            long timeSlot = Long.valueOf(map.get("timeSlot").toString());
            String timeSlotUnit = map.get("timeSlotUnit").toString();
            String timeSlotSymbol = map.get("timeSlotSymbol").toString();
            boolean b3 = validateTime(slotsMillis, timeSlot, timeSlotUnit, timeSlotSymbol);
            if (!b3) {
                return "";
            }
        }
        //已分配 VCore 的秒数
        String allocatedVcoreSeconds = parameterModel.getAllocatedVcoreSeconds();
        if (StringUtils.isNotBlank(allocatedVcoreSeconds) && "null" != allocatedVcoreSeconds && StringUtils.isNotBlank(map.get("vcore").toString())) {
            long allocatedVcoreSeconds1 = Long.valueOf(allocatedVcoreSeconds);
            long vcore = Long.valueOf(map.get("vcore").toString());
            String vcoreSymbol = map.get("vcoreSymbol").toString();
            boolean b = validateValue(allocatedVcoreSeconds1, vcore, vcoreSymbol);
            if (!b) {
                return "";
            }
        }
        //已分配内存的秒数
        String allocatedMemorySeconds = parameterModel.getAllocatedMemorySeconds();
        if (StringUtils.isNotBlank(allocatedMemorySeconds) && "null" != allocatedMemorySeconds && StringUtils.isNotBlank(map.get("memory").toString())) {
            long allocatedMemorySeconds1 = Long.valueOf(allocatedMemorySeconds);
            long memory = Long.valueOf(map.get("memory").toString());
            String memorySymbol = map.get("memorySymbol").toString();
            boolean b = validateValue(allocatedMemorySeconds1, memory, memorySymbol);
            if (!b) {
                return "";
            }
        }
        //读取的 HDFS 字节
        String a4 = (String) parseObject.get("hdfs_bytes_read");
        if (StringUtils.isNotBlank(a4) && "null" != a4 && StringUtils.isNotBlank(map.get("byte").toString())) {
            long hdfsBytesRead = Long.valueOf(a4);
            long byte1 = Long.valueOf(map.get("byte").toString());
            String byteUnit = map.get("byteUnit").toString();
            String byteSymbol = map.get("byteSymbol").toString();
            boolean b = validateByte(hdfsBytesRead, byte1, byteUnit, byteSymbol);
            if (!b) {
                return "";
            }
        }
        //写入的文件字节
        String a5 = (String) parseObject.get("file_bytes_written");
        if (StringUtils.isNotBlank(a5) && "null" != a5 && StringUtils.isNotBlank(map.get("write").toString())) {
            long fileBytesWritten = Long.valueOf(a5);
            long write = Long.valueOf(map.get("write").toString());
            String writeUnit = map.get("writeUnit").toString();
            String writeSymbol = map.get("writeSymbol").toString();
            boolean b = validateByte(fileBytesWritten, write, writeUnit, writeSymbol);
            if (!b) {
                return "";
            }
        }
        //总任务
        String a6 = (String) parseObject.get("total_launched_tasks");
        if (StringUtils.isNotBlank(a6) && "null" != a6 && StringUtils.isNotBlank(map.get("task").toString())) {
            long total = Long.valueOf(a6);
            long task = Long.valueOf(map.get("task").toString());
            String taskSymbol = map.get("taskSymbol").toString();
            boolean b = validateValue(total, task, taskSymbol);
            if (!b) {
                return "";
            }
        }
        return model.getModelName();
    }

    public String validateFlinkAndSparkModelName(YarnParameterModel parameterModel, ConfigureModel model) {
        String param = model.getParam();
        Map map = JSON.parseObject(param, Map.class);
        Map parseObject = JSON.parseObject(parameterModel.getOtherParameterData(), Map.class);
        //持续时间
        log.info("11111111111111111111111111111111111");
        if (StringUtils.isNotBlank(parameterModel.getElapsedTime()) && StringUtils.isNotBlank(map.get("continue").toString())) {
            long durationMillis = Long.valueOf(parameterModel.getElapsedTime());
            long aContinue = Long.valueOf(map.get("continue").toString());
            String continueUnit = map.get("continueUnit").toString();
            String continueSymbol = map.get("continueSymbol").toString();
            boolean b1 = validateTime(durationMillis, aContinue, continueUnit, continueSymbol);
            if (!b1) {
                return "";
            }
        }
        //已分配 VCore 的秒数
        log.info("222222222222222222222222222222");
        String allocatedVcoreSeconds = parameterModel.getAllocatedVcoreSeconds();
        if (StringUtils.isNotBlank(allocatedVcoreSeconds) && "null" != allocatedVcoreSeconds && StringUtils.isNotBlank(map.get("vcore").toString())) {
            long allocatedVcoreSeconds1 = Long.valueOf(allocatedVcoreSeconds);
            long vcore = Long.valueOf(map.get("vcore").toString());
            String vcoreSymbol = map.get("vcoreSymbol").toString();
            boolean b = validateValue(allocatedVcoreSeconds1, vcore, vcoreSymbol);
            if (!b) {
                return "";
            }
        }
        //已分配内存的秒数
        log.info("3333333333333333333333333");
        String allocatedMemorySeconds = parameterModel.getAllocatedMemorySeconds();
        if (StringUtils.isNotBlank(allocatedMemorySeconds) && "null" != allocatedMemorySeconds && StringUtils.isNotBlank(map.get("memory").toString())) {
            long allocatedMemorySeconds1 = Long.valueOf(allocatedMemorySeconds);
            long memory = Long.valueOf(map.get("memory").toString());
            String memorySymbol = map.get("memorySymbol").toString();
            boolean b = validateValue(allocatedMemorySeconds1, memory, memorySymbol);
            if (!b) {
                return "";
            }
        }
        //已分配的 VCore 数
        log.info("44444444444444444444444444444");
        String allocatedVCores = (String) parseObject.get("allocatedVCores");
        if (StringUtils.isNotBlank(allocatedVCores) && "null" != allocatedVCores && StringUtils.isNotBlank(map.get("allocationVcore").toString())) {
            long allocatedVcoreSeconds1 = Long.valueOf(allocatedVCores);
            long vcore = Long.valueOf(map.get("allocationVcore").toString());
            String vcoreSymbol = map.get("allocationVcoreSymbol").toString();
            boolean b = validateValue(allocatedVcoreSeconds1, vcore, vcoreSymbol);
            if (!b) {
                return "";
            }
        }
        //已分配内存的秒数
        log.info("555555555555555555555555555");
        String allocatedMB = (String) parseObject.get("allocatedMB");
        if (StringUtils.isNotBlank(allocatedMB) && "null" != allocatedMB && StringUtils.isNotBlank(map.get("allocationMemory").toString())) {
            long allocatedMemorySeconds1 = Long.valueOf(allocatedMB);
            long memory = Long.valueOf(map.get("allocationMemory").toString());
            String memorySymbol = map.get("allocationMemorySymbol").toString();
            boolean b = validateValue(allocatedMemorySeconds1, memory, memorySymbol);
            if (!b) {
                return "";
            }
        }
        //正在运行的容器数
        log.info("6666666666666666666666666666");
        String runningContainers = (String) parseObject.get("runningContainers");
        if (StringUtils.isNotBlank(runningContainers) && "null" != runningContainers && StringUtils.isNotBlank(map.get("container").toString())) {
            long runningContainers1 = Long.valueOf(runningContainers);
            long container = Long.valueOf(map.get("container").toString());
            String containerSymbol = map.get("containerSymbol").toString();
            boolean b = validateValue(runningContainers1, container, containerSymbol);
            if (!b) {
                return "";
            }
        }
        return model.getModelName();
    }

    public void getFlinkParam(Map<String, String> map, String startDate) {
        try {
            log.info("获取flink参数开始********************************");
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/yarnApplications?filter=(application_type='Apache Flink')&from={startTime}";
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmFlag = map.get("cmFlag");
            String cmName = map.get("cmName");
            String clusterFlag = map.get("clusterFlag");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            //时间为当前时间前10分钟
            String iso86Time = getISO86Time(startDate);
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName)
                    .replace("{startTime}", iso86Time);
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                JSONObject obj = JSONObject.parseObject(objectMap.get("body").toString());
                JSONArray jsonArray = obj.getJSONArray("applications");
                List<YarnParameterModel> list = new ArrayList<>();
                if (jsonArray != null && jsonArray.size() > 0) {
                    List<ConfigureModel> flinkModels = statementAuditDao.selectModelConfig("FLINK","FINISHED");
                    List<ConfigureModel> flinkModelsError = statementAuditDao.selectModelConfig("FLINK","FAILED");
                    List<ConfigureModel> flinkModelsPending = statementAuditDao.selectModelConfig("FLINK","PENDING");
                    int flinkCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        log.info("flink parseObject:" + parseObject);
                        if("FAILED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : flinkModelsError) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "1");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("FINISHED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : flinkModels) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "1");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("NEW".equalsIgnoreCase(parseObject.getString("state")) ||
                                "NEW_SAVING".equalsIgnoreCase(parseObject.getString("state")) ||
                                "SUBMITTED".equalsIgnoreCase(parseObject.getString("state")) ||
                                "ACCEPTED".equalsIgnoreCase(parseObject.getString("state"))) {
                            for (ConfigureModel model : flinkModelsPending) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "1");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                    flinkCount++;
                                }
                            }
                        }
                    }
                    log.info("flink list大小" + list.size());
                    if (list.size() > 0) {
                        insertMonthYarnParameterData(list);
                        //等待任务数量告警
                        if (flinkCount > 0){
                            String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+yarnPublishTheme+"：FLINK任务等待数 "+ flinkCount;
                            //微信、钉钉告警，然后添加进告警日志表
                            cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,yarnPublishTheme);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取yarn语句flink任务参数异常：{}" + e);
        }
    }

    public void copyParam(String cmFlag, String cmName, String clusterFlag, String clusterName,
                          YarnParameterModel parameterModel, JSONObject parseObject, String type) throws ParseException {
        parameterModel.setCmFlag(cmFlag);
        parameterModel.setCmName(cmName);
        parameterModel.setClusterFlag(clusterFlag);
        parameterModel.setClusterName(clusterName);
        parameterModel.setApplicationId(parseObject.getString("applicationId"));
        if ("1".equals(type)) {
            parameterModel.setApplicationType("Apache Flink");
        } else if ("2".equals(type)) {
            parameterModel.setApplicationType("SPARK");
        } else if ("3".equals(type)) {
            parameterModel.setApplicationType("TEZ");
        }

        parameterModel.setName(parseObject.get("name").toString().replace("\n", " "));
        parameterModel.setPool(parseObject.getString("pool"));
        parameterModel.setUser(parseObject.getString("user"));
        parameterModel.setState(parseObject.getString("state"));
        parameterModel.setAllocatedMemorySeconds(parseObject.get("allocatedMemorySeconds").toString());
        parameterModel.setAllocatedVcoreSeconds(parseObject.get("allocatedVcoreSeconds").toString());
        Map<String, String> outherParamMap = new HashMap<>();
        outherParamMap.put("allocatedMB", parseObject.get("allocatedMB").toString());
        outherParamMap.put("allocatedVCores", parseObject.get("allocatedVCores").toString());
        outherParamMap.put("runningContainers", parseObject.get("runningContainers").toString());
        outherParamMap.put("progress", parseObject.get("progress").toString());
        parameterModel.setOtherParameterData(JSONArray.toJSON(outherParamMap).toString());
        parameterModel.setStartTime(getBjTime(parseObject.getString("startTime")));
        Date dat = new Date();
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        if (parseObject.get("endTime") != null) {
            parameterModel.setEndTime(getBjTime(parseObject.getString("endTime")));
            long elapsedTime = getElapsedTime(parameterModel.getStartTime(), parameterModel.getEndTime());
            parameterModel.setElapsedTime(String.valueOf(elapsedTime));
        } else {
            long elapsedTime = getElapsedTime(parameterModel.getStartTime(), sf.format(dat));
            parameterModel.setElapsedTime(String.valueOf(elapsedTime));
        }
        String date = sf.format(dat);
        parameterModel.setCreateTime(date);
    }

    public void getSparkParam(Map<String, String> map, String startDate) {
        try {
            log.info("获取spark参数开始**********************************");
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/yarnApplications?filter=(application_type=SPARK)&from={startTime}";
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmFlag = map.get("cmFlag");
            String cmName = map.get("cmName");
            String clusterFlag = map.get("clusterFlag");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            //时间为当前时间前10分钟
            String iso86Time = getISO86Time(startDate);
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName)
                    .replace("{startTime}", iso86Time);
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                JSONObject obj = JSONObject.parseObject(objectMap.get("body").toString());
                JSONArray jsonArray = obj.getJSONArray("applications");
                List<YarnParameterModel> list = new ArrayList<>();
                if (jsonArray != null && jsonArray.size() > 0) {
                    List<ConfigureModel> sparkModels = statementAuditDao.selectModelConfig("SPARK","FINISHED");
                    List<ConfigureModel> sparkModelsError = statementAuditDao.selectModelConfig("SPARK","FAILED");
                    List<ConfigureModel> sparkModelsPending = statementAuditDao.selectModelConfig("SPARK","PENDING");
                    //获取等待中任务数量
                    int sparkCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        log.info("spark parseObject:" + parseObject);
                        if("FAILED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : sparkModelsError) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "2");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("FINISHED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : sparkModels) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "2");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("NEW".equalsIgnoreCase(parseObject.getString("state")) ||
                                "NEW_SAVING".equalsIgnoreCase(parseObject.getString("state")) ||
                                "SUBMITTED".equalsIgnoreCase(parseObject.getString("state")) ||
                                "ACCEPTED".equalsIgnoreCase(parseObject.getString("state"))) {
                            for (ConfigureModel model : sparkModelsPending) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "2");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                    sparkCount++;
                                }
                            }
                        }
                    }
                    log.info("spark list大小:" + list.size());
                    if (list.size() > 0) {
                        insertMonthYarnParameterData(list);
                        //等待任务数量告警
                        if (sparkCount > 0){
                            String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+yarnPublishTheme+"：SPARK任务等待数 "+ sparkCount;
                            //微信、钉钉告警，然后添加进告警日志表
                            cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,yarnPublishTheme);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取yarn语句spark任务参数异常：{}" + e);
        }
    }

    public void getTezParam(Map<String, String> map, String startDate) {
        try {
            log.info("获取TEZ参数开始**********************************");
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/yarnApplications?filter=(application_type=TEZ)&from={startTime}";
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmFlag = map.get("cmFlag");
            String cmName = map.get("cmName");
            String clusterFlag = map.get("clusterFlag");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            //时间为当前时间前10分钟
            String iso86Time = getISO86Time(startDate);
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName)
                    .replace("{startTime}", iso86Time);
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                JSONObject obj = JSONObject.parseObject(objectMap.get("body").toString());
                JSONArray jsonArray = obj.getJSONArray("applications");
                List<YarnParameterModel> list = new ArrayList<>();
                if (jsonArray != null && jsonArray.size() > 0) {
                    List<ConfigureModel> TezModels = statementAuditDao.selectModelConfig("TEZ","FINISHED");
                    List<ConfigureModel> TezModelsError = statementAuditDao.selectModelConfig("TEZ","FAILED");
                    List<ConfigureModel> TezModelsPending = statementAuditDao.selectModelConfig("TEZ","PENDING");
                    //获取等待中任务数量
                    int TezCount = 0;
                    for (int i = 0; i < jsonArray.size(); i++) {
                        JSONObject parseObject = jsonArray.getJSONObject(i);
                        log.info("TEZ parseObject:" + parseObject);
                        if("FAILED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : TezModelsError) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "3");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("FINISHED".equals(parseObject.getString("state"))) {
                            for (ConfigureModel model : TezModels) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "3");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                }
                            }
                        }
                        if("NEW".equalsIgnoreCase(parseObject.getString("state")) ||
                                "NEW_SAVING".equalsIgnoreCase(parseObject.getString("state")) ||
                                "SUBMITTED".equalsIgnoreCase(parseObject.getString("state")) ||
                                "ACCEPTED".equalsIgnoreCase(parseObject.getString("state"))) {
                            for (ConfigureModel model : TezModelsPending) {
                                YarnParameterModel parameterModel = new YarnParameterModel();
                                copyParam(cmFlag, cmName, clusterFlag, clusterName, parameterModel, parseObject, "3");
                                String modelName = validateFlinkAndSparkModelName(parameterModel, model);
                                if (StringUtils.isNotBlank(modelName)) {
                                    parameterModel.setModelName(modelName);
                                    list.add(parameterModel);
                                    TezCount++;
                                }
                            }
                        }
                    }
                    log.info("TEZ list大小:" + list.size());
                    if (list.size() > 0) {
                        insertMonthYarnParameterData(list);
                        //等待任务数量告警
                        if (TezCount > 0){
                            String warnAlert = "<"+cmName+">的<"+clusterName+">集群，"+yarnPublishTheme+"：SPARK任务等待数 "+ TezCount;
                            //微信、钉钉告警，然后添加进告警日志表
                            cmTargetParamService.warnWxDingDingAndLog(cmFlag,cmName,clusterFlag,clusterName,serviceName,warnAlert,yarnPublishTheme);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("获取yarn语句tez任务参数异常：{}" + e);
        }
    }

    //Yarn语句审计保留历史记录（2年）数据表按照月份表
    public void insertMonthYarnParameterData(List<YarnParameterModel> list) {
        try {
            int pointsDataLimit = 200;
            int listSize = list.size();
            int maxSize = listSize - 1;
            List<YarnParameterModel> newList = new ArrayList<>();
            //获取当前月表名
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMM");
            String name = yarnTableName + "_" + sdf.format(new Date());
            logger.info("当前月表名："+name);
            for (int i = 0; i < listSize; i++) {
                newList.add(list.get(i));
                if (pointsDataLimit == newList.size() || i == maxSize) {
                    Map map = new HashMap();
                    map.put("yarnList",newList);
                    map.put("tableName",name);
                    statementAuditDao.insertMonthYarnParameterData(map);
                    newList.clear();
                }
            }
        } catch (Exception e) {
            log.error("yarn任务参数插入yarn表异常：{}" + e);
        }
    }

    public String getImpalaTableInfo(String sql) {
        System.out.println(sql);
        String s = "";
        if (StringUtils.isEmpty(sql)) {
            return "";
        }
        try {
            String format = SQLUtils.format(sql, DbType.mysql);
            SQLStatementParser sqlStatementParser = SQLParserUtils.createSQLStatementParser(format, DbType.mysql);
            SQLStatement statement = sqlStatementParser.parseStatement();
            MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor();
            statement.accept(visitor);
            s = visitor.getTables().toString();
        } catch (Exception e) {
            log.error("解析impala sql表名失败：" + e);
            log.error("impala sql:" + sql);
            return "";
        }
        return s;
    }

    public String getMapreduceTableInfo(String sql) {
        System.out.println(sql);
        String s = "";
        if (StringUtils.isEmpty(sql)) {
            return "";
        }
        try {
            String format = SQLUtils.format(sql, DbType.hive);
            SQLStatementParser sqlStatementParser = SQLParserUtils.createSQLStatementParser(format, DbType.hive);
            SQLStatement statement = sqlStatementParser.parseStatement();
            MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor();
            statement.accept(visitor);
            s = visitor.getTables().toString();
        } catch (Exception e) {
            log.error("解析mapreduce sql表名失败：" + e);
            log.error("mapreduce sql:" + sql);
            return "";
        }
        return s;
    }

    public static String getISO86Time(String date) {
        if (StringUtils.isBlank(date)) {
            return "";
        }
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date newTime = null;
        try {
            newTime = format.parse(date);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        TimeZone tz = TimeZone.getTimeZone("GMT-01");
        DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
        df.setTimeZone(tz);
        return df.format(newTime);
    }

    public static String getBjTime(String ISOdate) {
        if (StringUtils.isBlank(ISOdate)) {
            return "";
        }
        DateTimeFormatter dtf1 = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
        DateTime dt = dtf1.parseDateTime(ISOdate);
        DateTimeFormatter dtf2 = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
        return dt.toString(dtf2);
    }

    /**
     * 查询impala任务同类语句并发数
     */
    public int selectImpalaConcurrency(String queryType,String tableInfo, String createTime) throws Exception {
        Set set = new HashSet();
        String startTime = getThreeDayByDate(createTime);
        //获取时间范围内所有月份表
        List<String> stringList = connMysqlUtil.getMonthBetweenDate(startTime, createTime);
        String tableName = impalaTableName + "_" + stringList.get(0);
        List<String> tableNameList = new ArrayList<>();
        if (stringList.size()>1){
            for (int i=1;i<stringList.size();i++) {
                String name = impalaTableName + "_" + stringList.get(i);
                tableNameList.add(name);
            }
        }
        List<Map> mapList = statementAuditDao.selectImpalaConcurrency(queryType,tableInfo, startTime, createTime,tableName,tableNameList);
        for (Map map : mapList) {
            Clob clob = (Clob) map.get("otherParameterDatafrom");
            Map dataMap = JSONObject.parseObject(clob.getSubString(1, (int) clob.length()), Map.class);
            String info = (String) dataMap.get("tableInfo");
            if (info.equals(tableInfo)){
                set.add(map.get("queryId"));
            }
        }
        return set.size();
    }

    /**
     * 查询mapreduce任务同类语句并发数
     */
    public int selectMapreduceConcurrency(String tableInfo, String createTime) throws Exception {
        Set set = new HashSet();
        String startTime = getThreeDayByDate(createTime);
        //获取时间范围内所有月份表
        List<String> stringList = connMysqlUtil.getMonthBetweenDate(startTime, createTime);
        String tableName = yarnTableName + "_" + stringList.get(0);
        List<String> tableNameList = new ArrayList<>();
        if (stringList.size()>1){
            for (int i=1;i<stringList.size();i++) {
                String name = yarnTableName + "_" + stringList.get(i);
                tableNameList.add(name);
            }
        }
        List<Map> mapList = statementAuditDao.selectMapreduceConcurrency(tableInfo, startTime, createTime,tableName,tableNameList);
        for (Map map : mapList) {
            Clob clob = (Clob) map.get("otherParameterDatafrom");
            Map dataMap = JSONObject.parseObject(clob.getSubString(1, (int) clob.length()), Map.class);
            String info = (String) dataMap.get("tableInfo");
            if (info.equals(tableInfo)){
                set.add(map.get("applicationId"));
            }
        }
        return set.size();
    }

    public String getThreeDayByDate(String date) throws ParseException {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        long beginTime = sf.parse(date).getTime();
        long intervalTime = -3*24*60*60*1000;
        long lastTime = beginTime + intervalTime;
        Date date1 = new Date(lastTime);
        return sf.format(date1);
    }

    public Object getImpalaProfileById(String cmFlag, String clusterFlag, String id) {
        List<Map<String, String>> monitorReqList = cmSwitchService.getCmConfigInfo(cmFlag, clusterFlag, "impala");
        for (Map<String, String> map : monitorReqList) {
            String apiUrl = "{url}/clusters/{clusterFlag}/services/{serviceName}/impalaQueries/" + id;
            String url = map.get("url");
            String user = map.get("user");
            String password = map.get("password");
            String cmName = map.get("cmName");
            String clusterName = map.get("clusterName");
            String serviceName = map.get("serviceName");
            String replaceUrl = apiUrl.replace("{url}", url)
                    .replace("{clusterFlag}", clusterFlag)
                    .replace("{serviceName}", serviceName);
            Map<String, Object> objectMap = cmSwitchService.sendRequest(replaceUrl, new HashMap<>(), HttpMethod.GET, user, password);
            int code = Integer.parseInt(objectMap.get("code").toString());
            if (200 == code) {
                String body = objectMap.get("body").toString();
                Map<String, String> jsonMap = JSON.parseObject(body, new TypeReference<HashMap<String, String>>() {
                });
                return jsonMap.get("details");
            }
        }
        return null;
    }

    /**
     * 按月创建 tb_impala_parameter、tb_yarn_parameter月份表
     * 每月月底执行
     */
    public void createTableMonth() {
        try {
            //获取下个月表名
            String impalaName = impalaTableName + "_" + connMysqlUtil.getNextMonthYear();
            String yarnName = yarnTableName + "_" + connMysqlUtil.getNextMonthYear();
            //判断表是否存在
            if (!connMysqlUtil.selectTable(impalaName)){
                //获取建表文件的sql语句
                String createSql = connMysqlUtil.readSqlFile(impalaSqlPath);
                String sqlData = createSql.replace("{tableName}", impalaName);
                //建表
                connMysqlUtil.createTable(impalaName,sqlData);
            }
            if (!connMysqlUtil.selectTable(yarnName)){
                //获取建表文件的sql语句
                String createSql = connMysqlUtil.readSqlFile(yarnSqlPath);
                String sqlData = createSql.replace("{tableName}", yarnName);
                //建表
                connMysqlUtil.createTable(yarnName,sqlData);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("按月创建 tb_impala_parameter、tb_yarn_parameter月份表异常：{}" + e);
        }
    }

    /**
     * 删除2年前tb_impala_parameter、tb_yarn_parameter月份表
     * 月底执行
     */
    public void dropTableMonth() {
        try {
            //判断2年前月表是否存在
            String impalaNameYear = impalaTableName + "_" + connMysqlUtil.getYearBetweenDate();
            String yarnNameYear = yarnTableName + "_" + connMysqlUtil.getYearBetweenDate();
            if (connMysqlUtil.selectTable(impalaNameYear)){
                //删除2年前的月份表
                connMysqlUtil.deleteTable(impalaNameYear);
            }
            if (connMysqlUtil.selectTable(yarnNameYear)){
                connMysqlUtil.deleteTable(yarnNameYear);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("删除2年前tb_impala_parameter、tb_yarn_parameter月份表异常：{}" + e);
        }
    }
}
